package main

//
//import (
//	"encoding/json"
//	"fmt"
//	//flow "github.com/s8sg/goflow/flow/v1"
//	//goflow "github.com/s8sg/goflow/v1"
//	"math/rand"
//	"strconv"
//)
//
//func Input(data []byte, option map[string][]string) ([]byte, error) {
//	var input map[string]int
//	if err := json.Unmarshal(data, &input); err != nil {
//		return nil, err
//	}
//	outputInt := input["input"]
//
//	return []byte(strconv.Itoa(outputInt)), nil
//
//}
//
//func AddOne(data []byte, option map[string][]string) ([]byte, error) {
//	num, _ := strconv.Atoi(string(data))
//	outputInt := num + rand.Intn(10) + 1
//	fmt.Println("AddOne=", outputInt)
//
//	return []byte(strconv.Itoa(outputInt)), nil
//}
//
//func AddTwo(data []byte, option map[string][]string) ([]byte, error) {
//	num, _ := strconv.Atoi(string(data))
//	outputInt := num + rand.Intn(10) + 1
//	fmt.Println("AddTwo=", outputInt)
//
//	return []byte(strconv.Itoa(outputInt)), nil
//}
//
//func Aggregator(data []byte, option map[string][]string) ([]byte, error) {
//	fmt.Println("Aggregator = ", string(data))
//	return data, nil
//}
//
//// Expand10 扩大十倍
//func Expand10(data []byte, option map[string][]string) ([]byte, error) {
//	num, _ := strconv.Atoi(string(data))
//	outputInt := num * 10
//	fmt.Println("Expand10=", outputInt)
//
//	return []byte(strconv.Itoa(outputInt)), nil
//}
//
//// Expand10 扩大100倍
//func Expand100(data []byte, option map[string][]string) ([]byte, error) {
//	num, _ := strconv.Atoi(string(data))
//	outputInt := num * 100
//	fmt.Println("Expand10=", outputInt)
//
//	return []byte(strconv.Itoa(outputInt)), nil
//}
//
//func Output(data []byte, option map[string][]string) ([]byte, error) {
//
//	fmt.Println("data=", string(data))
//
//	return []byte("ok"), nil
//}
//
//func main() {
//	// 创建工作流
//
//	fs := goflow.FlowService{
//		Port:              8080,
//		RedisURL:          "127.0.0.1:6379",
//		WorkerConcurrency: 5,
//	}
//
//	fs.Register("myflow", Myflow)
//	fs.Register("complexflow", MyComplexFlow)
//
//	if err := fs.Start(); err != nil {
//		panic(any(err))
//	}
//}
//
//func MyComplexFlow(workflow *flow.Workflow, context *flow.Context) error {
//
//	//构建节点
//	dag := workflow.Dag()
//	dag.Node("input", Input)
//	dag.Node("add-one", AddOne)
//	dag.Node("add-two", AddTwo)
//
//	// 聚合
//	dag.Node("aggregator", Aggregator, flow.Aggregator(func(m map[string][]byte) ([]byte, error) {
//		a, _ := strconv.Atoi(string(m["add-one"]))
//		b, _ := strconv.Atoi(string(m["add-two"]))
//		num := a + b
//		fmt.Println("num:", num)
//
//		return []byte(strconv.Itoa(num)), nil
//	}))
//
//	branchs := dag.ConditionalBranch("judge", []string{"moreThan", "lessThan"}, func(bytes []byte) []string {
//		// 聚合后的结果
//		num, _ := strconv.Atoi(string(bytes))
//		fmt.Println("ConditionalBranch = ", num)
//		if num > 10 {
//			return []string{"moreThan"}
//		}
//		return []string{"lessThan"}
//	}, flow.Aggregator(func(m map[string][]byte) ([]byte, error) {
//		// 返回对应的结果
//		if v, ok := m["moreThan"]; ok {
//
//			return v, nil
//		}
//
//		if v, ok := m["lessThan"]; ok {
//			return v, nil
//		}
//
//		return nil, nil
//	}))
//
//	// 创建节点
//	branchs["moreThan"].Node("expand-10", Expand10)
//	branchs["lessThan"].Node("expand-100", Expand100)
//	dag.Node("output", Output)
//
//	// 构建依赖关系
//	dag.Edge("input", "add-one")
//	dag.Edge("input", "add-two")
//	// 聚合
//	dag.Edge("add-one", "aggregator")
//	dag.Edge("add-two", "aggregator")
//
//	dag.Edge("aggregator", "judge")
//	dag.Edge("judge", "output")
//
//	return nil
//
//}
//
//func Myflow(flow *flow.Workflow, context *flow.Context) error {
//
//	//构建节点
//	dag := flow.Dag()
//	dag.Node("input", Input)
//	dag.Node("add-one", AddOne)
//	dag.Node("add-two", AddTwo)
//	dag.Node("output", Output)
//	// 构建依赖关系
//	dag.Edge("input", "add-one")
//	dag.Edge("add-one", "add-two")
//	dag.Edge("add-two", "output")
//
//	return nil
//
//}
